package avicit.bdp.dcs.threshold.base;

import avicit.bdp.dcs.job.dto.JobDTO;
import avicit.bdp.dcs.task.dto.TaskDTO;
import avicit.bdp.dcs.threshold.utils.TestUtils;
import avicit.platform6.core.rest.msg.ResponseMsg;
import com.alibaba.fastjson2.JSON;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.springframework.http.HttpMethod;
import org.springframework.http.ResponseEntity;
import org.springframework.web.client.RestTemplate;

import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

import static org.junit.Assert.assertEquals;

/**
 * 增量同步任务测试用例
 */
public class DataSyncIncrementTest {
    private RestTemplate restTemplate = new RestTemplate();

    private String jobName = "数据增量同步门槛用例";

    private static String srcTble = "inc_src_" + System.currentTimeMillis();

    private static String dstTble = "inc_dst_" + System.currentTimeMillis();

    /**
     * 所有用例执行前准备工作
     */
    @BeforeClass
    public static void init() {
        System.out.println("数据增量同步门槛用例用例初始化...");
        TestUtils.insert2Mysql(srcTble);
    }

    /**
     * 单个测试用例运行前准备工作
     */
    @Before
    public void setUp() {
        // TODO:
    }

    /**
     * 单个测试用例运行后销毁工作
     */
    @After
    public void tearDown() {
        // TODO:
    }

    /**
     * 所有用例执行完成后准备工作
     */
    @AfterClass
    public static void destory() {
        System.out.println("数据增量同步门槛用例用例执行完成，开始清理...");
        TestUtils.deleteFromMysql(srcTble);
        TestUtils.deleteFromMysql(dstTble);
    }

    /***********************************************Begin 测试用例 ***********************************************/
    /**
     * 数据同步--增量同步
     *
     * @throws Exception
     */
    @Test(timeout = 5 * 60 * 1000)
    public void testDataSyncIncrement() throws Exception {
        // step1:保存数据源
        ResponseEntity<ResponseMsg> result = restTemplate.postForEntity(TestUtils.HOSTPORT
                + "/api/bdp/dcs/job/JobRest/saveDataSource", TestUtils.buildSaveDataSourceParas(2,
                jobName, TestUtils.MYSQL_SRC_DATASOURCE_ID, TestUtils.MYSQL_DST_DATASOURCE_ID, 0), ResponseMsg.class);
        TestUtils.handleFailedRequest(result);
        assertEquals("200", result.getBody().getRetCode());
        // 获取创建的JobId
        ResponseMsg responseMsg = result.getBody();
        Map<String, String> jobMap = (Map<String, String>) responseMsg.getResponseBody();
        String jobId = jobMap.get("jobId");

        try {
            // step2:保存表映射关系
            result = restTemplate.postForEntity(TestUtils.HOSTPORT + "/api/bdp/dcs/job/JobRest/saveTableMapping",
                    TestUtils.buildSaveTblMapping(2, jobId, srcTble, dstTble), ResponseMsg.class);
            TestUtils.handleFailedRequest(result);
            assertEquals("200", result.getBody().getRetCode());

            // step3:根据JobId名称获取taskId
            result = restTemplate.exchange(TestUtils.HOSTPORT + "/api/bdp/dcs/task/TaskRest/getTaskById?jobId="
                    + jobId + "&srcTableName=" + srcTble, HttpMethod.GET, TestUtils.buildGetHttpEntity(), ResponseMsg.class);
            TestUtils.handleFailedRequest(result);
            assertEquals("200", result.getBody().getRetCode());
            ResponseMsg body = result.getBody();
            LinkedHashMap<String, Object> map = (LinkedHashMap<String, Object>) body.getResponseBody();
            TaskDTO taskDTO = JSON.parseObject(JSON.toJSONString(map), TaskDTO.class);
            System.out.println(taskDTO.getId());
            String taskId = taskDTO.getId();

            // step4:保存增量同步配置信息(首先根据JobId获取TaskId)
            result = restTemplate.postForEntity(TestUtils.HOSTPORT + "/api/bdp/dcs/task/TaskRest/updateTask",
                    TestUtils.buildUpdateTask(2, taskId, srcTble, dstTble), ResponseMsg.class);
            TestUtils.handleFailedRequest(result);
            assertEquals("200", result.getBody().getRetCode());

            // step5:任务配置检查
            result = restTemplate.postForEntity(TestUtils.HOSTPORT + "/api/bdp/dcs/job/JobRest/checkTaskConfig",
                    TestUtils.buildCheckTaskConfig(jobId), ResponseMsg.class);
            TestUtils.handleFailedRequest(result);
            assertEquals("200", result.getBody().getRetCode());

            // step6:预检查
            result = restTemplate.postForEntity(TestUtils.HOSTPORT + "/api/bdp/dcs/precheck/PrecheckRest" +
                            "/executePrecheck",
                    TestUtils.buildPreChecks(jobId), ResponseMsg.class);
            TestUtils.handleFailedRequest(result);
            assertEquals("200", result.getBody().getRetCode());

            // step7:更新Job状态
            result = restTemplate.postForEntity(TestUtils.HOSTPORT + "/api/bdp/dcs/job/JobRest/updateJobStatus",
                    TestUtils.buildUpdateJobStatus(jobId), ResponseMsg.class);
            TestUtils.handleFailedRequest(result);
            assertEquals("200", result.getBody().getRetCode());

            // step8:启用/禁用Job
            //        buildReleaseJob();
            //        result = restTemplate.exchange(HOSTPORT + "/api/bdp/dcs/job/JobRest/releaseJob?jobId="+JOBID+"&enable=1",
            //                HttpMethod.GET, httpReleaseJobEntity, ResponseMsg.class);
            //        assertEquals("200", result.getBody().getRetCode());

            // step9:启动任务
            // testUnit表中新增一条数据
            TestUtils.addData(srcTble);
            result = restTemplate.postForEntity(TestUtils.HOSTPORT + "/api/bdp/dcs/job/JobRest/startJobByIds?jobIds=" + jobId,
                    TestUtils.buildStartJob(jobId), ResponseMsg.class);
            TestUtils.handleFailedRequest(result);
            assertEquals("200", result.getBody().getRetCode());

            // step10:获取Job列表
            // 等待DS启动任务成功
            Thread.sleep(30 * 1000);
            result = restTemplate.exchange(TestUtils.HOSTPORT + "/api/bdp/dcs/job/JobRest/getJobList" +
                            "?jobStatus&sort&name=数据增量同步门槛用例&jobType=2&currentPage=1&pageSize=10",
                    HttpMethod.GET, TestUtils.buildGetHttpEntity(), ResponseMsg.class);
            body = result.getBody();
            map = (LinkedHashMap<String, Object>) body.getResponseBody();
            List<Object> finalBody = (List<Object>) map.get("result");
            for (Object obj : finalBody) {
                LinkedHashMap<String, String> jobObj = (LinkedHashMap<String, String>) obj;
                JobDTO jobInfo = JSON.parseObject(JSON.toJSONString(jobObj), JobDTO.class);
                // 判断任务是否启动成功；
                if (jobInfo.getId().equals(jobId)) {
                    System.out.println("数据增量同步门槛用例任务状态:" + jobInfo.getJobStatus());

                    // 判断任务是否同步完成;
                    assertEquals(7 == jobInfo.getJobStatus(), true);

                    // 判断同步速度;（备注：源表中只有一行数据，大小固定）
                    assertEquals(jobInfo.getReadSucceedBytes() == 10, true);

                    break;
                }
            }

            // step11:查看日志信息
            result = restTemplate.exchange(TestUtils.HOSTPORT + "/api/bdp/dcs/log/LogRest/getLogByTaskId" +
                    "?taskInstanceId=&skipLineNum=0" +
                    "&limit=100&jobId=" + jobId, HttpMethod.GET, TestUtils.buildGetHttpEntity(), ResponseMsg.class);
            TestUtils.handleFailedRequest(result);
            assertEquals("200", result.getBody().getRetCode());

            // step12:根据JobId名称获取taskId
            result = restTemplate.exchange(TestUtils.HOSTPORT + "/api/bdp/dcs/task/TaskRest/getTaskById?jobId="
                    + jobId + "&srcTableName=" + srcTble, HttpMethod.GET, TestUtils.buildGetHttpEntity(), ResponseMsg.class);
            TestUtils.handleFailedRequest(result);
            assertEquals("200", result.getBody().getRetCode());
            body = result.getBody();
            map = (LinkedHashMap<String, Object>) body.getResponseBody();
            taskDTO = JSON.parseObject(JSON.toJSONString(map), TaskDTO.class);
            System.out.println(taskDTO.getId());
            taskId = taskDTO.getId();
        } finally {
            // step13: 删除job
            result = restTemplate.exchange(TestUtils.HOSTPORT + "/api/bdp/dcs/job/JobRest/deleteJobByIds",
                    HttpMethod.DELETE,
                    TestUtils.buildDeleteJob(jobId),
                    ResponseMsg.class);
            TestUtils.handleFailedRequest(result);
            assertEquals("200", result.getBody().getRetCode());
        }
    }

}
